package com.example.influxdb;

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.beans.factory.annotation.Value;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @program: thingsboard
 * @description:
 * @author:fanzhihui
 * @create: 2022-04-07 09:35
 **/

public class InfluxDbUtils {
    @Value("${spring.influx.user:''}")
    private String userName;
    @Value("${spring.influx.password:''}")
    private String password;
    @Value("${spring.influx.url:''}")
    private String url;
    @Value("${spring.influx.database:''}")
    public String database;
    private String retentionPolicy;
    // InfluxDB实例
    private InfluxDB influxDB;

    Logger logger = LoggerFactory.getLogger(InfluxDbUtils.class);

    // 数据保存策略
    public static String policyNamePix = "autogen";

    public InfluxDbUtils(String userName, String password, String url, String database,
                         String retentionPolicy) {
        this.userName = userName;
        this.password = password;
        this.url = url;
        this.database = database;
        this.retentionPolicy = retentionPolicy == null || "".equals(retentionPolicy) ? policyNamePix : retentionPolicy;
        this.influxDB = influxDbBuild();
    }

    /**
     * 连接数据库 ，若不存在则创建
     *
     * @return influxDb实例
     */
    private InfluxDB influxDbBuild() {
        if (influxDB == null) {
            influxDB = InfluxDBFactory.connect(url, userName, password);
        }
        try {
            createDB(database);
            influxDB.setDatabase(database);
        } catch (Exception e) {
            logger.error("create influx db failed, error: {}", e.getMessage());
        } finally {
            influxDB.setRetentionPolicy(retentionPolicy);
        }
        influxDB.setLogLevel(InfluxDB.LogLevel.BASIC);
        return influxDB;
    }

    /****
     *  创建数据库
     * @param database
     */
    private void createDB(String database) {
        influxDB.query(new Query("CREATE DATABASE " + database));
    }

    /**
     * 删除数据库
     *
     * @param database
     */
    public void dropDatabase(String database) {
        influxDB.query(new Query("DROP DATABASE " + database, ""));
    }

    /**
     * 测试连接是否正常
     *
     * @return true 正常
     */
    public boolean ping() {
        boolean isConnected = false;
        Pong pong;
        try {
            pong = influxDB.ping();
            if (pong != null) {
                isConnected = true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return isConnected;
    }

    /**
     * 创建自定义保留策略
     *
     * @param policyName  策略名
     * @param duration    保存天数
     * @param replication 保存副本数量
     * @param isDefault   是否设为默认保留策略
     */
    public void createRetentionPolicy(String policyName, String duration, int replication, Boolean isDefault) {
        String sql = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s ", policyName,
                database, duration, replication);
        if (isDefault) {
            sql = sql + " DEFAULT";
        }
        this.query(sql); //influxDB.query(new Query(sql));
    }

    /**
     * 创建默认的保留策略
     *
     * @note 策略名：default，保存天数：30天，保存副本数量：1
     * 设为默认保留策略
     */
    public void createDefaultRetentionPolicy() {
        String command = String.format("CREATE RETENTION POLICY \"%s\" ON \"%s\" DURATION %s REPLICATION %s DEFAULT",
                "default", database, "30d", 1);
        this.query(command);
    }

    /**
     * 删除保留策略
     */
    public void dropRetentionPolicy() {
        this.dropRetentionPolicy(database, retentionPolicy);
    }

    public void dropRetentionPolicy(String database, String retentionPolicy) {
        String sql = String.format("DROP RETENTION POLICY %s ON %s", retentionPolicy, database);
        this.query(sql);
    }

    /**
     * 查询
     *
     * @param command 执行语句
     * @return
     */
    public QueryResult query(String command) {
        return influxDB.query(new Query(command));
    }


    /**
     * 插入数据参数
     *
     * @param measurement 表名
     * @param tags        tags域
     * @param fields      值
     * @param time        时间
     * @return
     */
    public Point pointBuilder(String measurement,
                              Map<String, String> tags,
                              Map<String, Object> fields,
                              long time) {
        Point point = Point.measurement(measurement).time(time, TimeUnit.MILLISECONDS).tag(tags).fields(fields).build();
        return point;
    }

    /**
     * 插入数据
     *
     * @param point1
     */
    public void insert(Point point1) {
        influxDB.write(point1);
    }

    /**
     * 插入
     *
     * @param measurement 表
     * @param tags        标签
     * @param fields      字段
     */
    public void insert(String measurement, Map<String, String> tags, Map<String, Object> fields, long time,
                       TimeUnit timeUnit) {
        Point.Builder builder = Point.measurement(measurement);
        builder.tag(tags);
        builder.fields(fields);
        if (0 != time) {
            builder.time(time, timeUnit);
        }
        influxDB.write(database, retentionPolicy, builder.build());
    }

    /**
     * 批量写入测点
     *
     * @param batchPoints
     */
    public void batchInsert(BatchPoints batchPoints) {
        influxDB.write(batchPoints);
    }

    /**
     * 批量写入数据
     *
     * @param database        数据库
     * @param retentionPolicy 保存策略
     * @param consistency     一致性
     * @param records         要保存的数据（调用BatchPoints.lineProtocol()可得到一条record）
     */
    public void batchInsert(final String database, final String retentionPolicy, final InfluxDB.ConsistencyLevel consistency,
                            final List<String> records) {
        influxDB.write(database, retentionPolicy, consistency, records);
    }

    /**
     * 删除
     *
     * @param command 删除语句
     * @return 返回错误信息
     */
    public String deleteMeasurementData(String command) {
        QueryResult result = influxDB.query(new Query(command, database));
        return result.getError();
    }

    /**
     * 关闭数据库
     */
    public void close() {
        influxDB.close();
    }

    public String getQueryCondition(Map<String, String> dimensionMap, LocalDateTime beginTime, LocalDateTime endTime) {
        String condition = "";
        for (String key : dimensionMap.keySet()) {
            System.out.println("key= " + key + " and value= " + dimensionMap.get(key));
            condition = condition + key+ "='"+dimensionMap.get(key) +"'";
        }

        if (beginTime != null && endTime != null){
            condition = condition +", time between "+beginTime + endTime;
        }

        return condition;
    }

}
